Pythonでリトライ処理を簡単に追加できる「tenacity」を使ってみる

今回はPythonで簡単にリトライ処理を追加できる「tenacity」を使ってみます。デコレータ形式で簡単にリトライ処理を追加できるので便利です。

#Python

アシ

2024.04.30

tenacityについて

プログラムを書いていて、HTTPの通信などでリトライ処理を実装する機会は多いと思います。

今回はそんなリトライ処理を簡潔に書けるtenacityの使い方を説明します。

インストール

インストールはpipで可能です。

pip install tenacity

使い方

シンプルな例

import random
from tenacity import retry

@retry
def random_error():
    num = random.randint(0, 10)
    if num > 4:
        print(f"Error: num={num}")
        raise Exception("Error!")
    else:
        return f"Success: num={num}"

print(random_error())
# =>
# Error: num=9
# Success: num=1

retryデコレータを使って対象の関数をラップするだけでリトライ処理を実装できます。

例外が発生した場合にはリトライが行われ、何も例外が発生しなければ値が返ってきます。

指定した回数だけ繰り返す

from tenacity import retry, stop_after_attempt, RetryError

@retry(stop=stop_after_attempt(3))
def error():
    print('Error occured!')
    raise Exception('Error occured!')

try:
    error()
except RetryError:
    print('3 times failed')
# =>
# Error occured!
# Error occured!
# Error occured!
# 3 times failed

retryデコレータに引数を渡すことでリトライ処理の挙動を変更可能です。

stop_after_attemptでは指定した回数だけリトライを行うことができます。

指定した回数に達した後に例外が発生した場合はRetryErrorが発生します。

時間を置いてからリトライする

from datetime import datetime
from tenacity import retry, stop_after_attempt, wait_fixed, RetryError

@retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
def error():
    print(f'Error occured!: {datetime.now().strftime("%M:%S")}')
    raise Exception('Error occured!')

try:
    error()
except RetryError:
    print('3 times failed')
# =>
# Error occured!: 06:38
# Error occured!: 06:41
# Error occured!: 06:44
# 3 times failed

ここではwait_fixedを使って、次のリトライまで３秒待っています。

例えば、APIのレート制限に引っかかって例外が発生した場合などに便利です。

特定の例外が発生した場合にリトライをする

from datetime import datetime
from tenacity import retry, stop_after_attempt, retry_if_exception_type, RetryError

@retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(ValueError))
def error(num):
    if num == 0:
        print('0 is invalid')
        raise ValueError()
    raise Exception(f'Error occured!: num={num}')

try:
    error(0)
except RetryError:
    print('Value Error occured')
# =>
# 0 is invalid
# 0 is invalid
# 0 is invalid
# Value Error occured

try:
    error(1)
except Exception as e:
    print(e)
# =>
# Error occured!: num=1

retry_if_exception_typeでリトライする場合の例外クラスを指定することができます。

ここでは０が引数として渡された場合はValueErrorが発生します。

その場合はリトライ処理が行われます。

一方で１を引数として渡した場合はValueError以外が発生するのでリトライ処理は行われません。

async/awaitでの例外処理

import asyncio
from tenacity import retry, stop_after_attempt, RetryError

@retry(stop=stop_after_attempt(3))
async def error():
    await asyncio.sleep(3)
    print('Error occured!')
    raise Exception('Error occured!')

try:
    loop = asyncio.get_event_loop()
    loop.run_until_complete(error())
except RetryError:
    print('3 times failed')
# => 
# Error occured!
# Error occured!
# Error occured!
# 3 times failed

tenacityはasync関数にも対応しています。

他の関数と同じようにデコレータでラップすることでリトライ処理を追加できます。

リトライ処理にも失敗した場合には通常の関数と同様にRetryErrorが発生します。